import mysql.connector
from elasticsearch import Elasticsearch, helpers
from datetime import datetime
import logging
from typing import Dict, Any
import sys

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('sync.log'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

# 数据库配置
MYSQL_CONFIG = {
    'host': 'localhost',
    'user': 'journeo',
    'password': 'journeo',
    'database': 'journeo'
}

# Elasticsearch配置
ES_CONFIG = {
    'hosts': ['http://localhost:9200']
}


def connect_to_mysql():
    """建立MySQL连接"""
    try:
        connection = mysql.connector.connect(**MYSQL_CONFIG)
        logger.info("MySQL连接成功")
        return connection
    except Exception as e:
        logger.error(f"MySQL连接失败: {str(e)}")
        raise


def connect_to_elasticsearch():
    """建立Elasticsearch连接"""
    try:
        es = Elasticsearch(**ES_CONFIG)
        if not es.ping():
            raise ValueError("无法连接到Elasticsearch")
        logger.info("Elasticsearch连接成功")
        return es
    except Exception as e:
        logger.error(f"Elasticsearch连接失败: {str(e)}")
        raise


def safe_float(value: Any) -> float:
    """安全地转换为float类型"""
    if value is None:
        return None
    try:
        return float(value)
    except (ValueError, TypeError):
        return None


def format_datetime(dt: datetime) -> str:
    """格式化datetime为ISO格式字符串"""
    if dt is None:
        return None
    return dt.isoformat()


def transform_record(record: Dict[str, Any]) -> Dict[str, Any]:
    """转换记录格式"""
    return {
        'attractionId': record['AttractionId'],
        'regionId': record['RegionId'],
        'name': record['Name'],
        'description': record['Description'],
        'address': record['Address'],
        'image': record['Image'],
        'latitude': safe_float(record['Latitude']),
        'longitude': safe_float(record['Longitude']),
        'priceRange': record['PriceRange'],
        'openingHours': record['OpeningHours'],
        'attractionRating': safe_float(record['AttractionRating']),
        'recommendedVisitTime': record['RecommendedVisitTime'],
        'recommendedVisitSeason': record['RecommendedVisitSeason'],
    }


def fetch_data(connection):
    """获取MySQL数据"""
    cursor = connection.cursor(dictionary=True)
    try:
        query = """
        SELECT 
            AttractionId, RegionId, Name, Description, Address,
            Image, Latitude, Longitude, PriceRange, OpeningHours,
            AttractionRating, RecommendedVisitTime, RecommendedVisitSeason
        FROM attractions
        """
        cursor.execute(query)
        records = cursor.fetchall()
        logger.info(f"从MySQL获取了 {len(records)} 条记录")
        return records
    except Exception as e:
        logger.error(f"查询数据失败: {str(e)}")
        raise
    finally:
        cursor.close()


def generate_actions(records):
    """生成Elasticsearch批量操作数据"""
    for record in records:
        doc = transform_record(record)
        yield {
            '_index': 'attractions',
            '_id': str(doc['attractionId']),
            '_source': doc
        }


def sync_data():
    """执行数据同步"""
    mysql_conn = None
    try:
        # 建立连接
        mysql_conn = connect_to_mysql()
        es = connect_to_elasticsearch()

        # 获取数据
        records = fetch_data(mysql_conn)

        # 批量导入
        success, failed = helpers.bulk(
            es,
            generate_actions(records),
            stats_only=True,
            chunk_size=500,  # 每批次处理500条记录
            request_timeout=60  # 超时时间60秒
        )

        logger.info(f"同步完成: 成功 {success} 条, 失败 {failed} 条")

    except Exception as e:
        logger.error(f"同步过程出错: {str(e)}")
        raise
    finally:
        if mysql_conn:
            mysql_conn.close()


def main():
    """主函数"""
    try:
        logger.info("开始数据同步")
        sync_data()
        logger.info("数据同步完成")
    except Exception as e:
        logger.error(f"程序执行失败: {str(e)}")
        sys.exit(1)


if __name__ == "__main__":
    main()